package lsm

import (
	"fmt"
	"learning/gooop/etcd/raft/logger"
	"learning/gooop/etcd/raft/model"
	"learning/gooop/etcd/raft/roles"
	"learning/gooop/etcd/raft/rpc"
	"learning/gooop/etcd/raft/timeout"
	"sync"
	"time"
)

// tCandidateState presents a candidate node
type tCandidateState struct {
	model.TEventDrivenModel

	context    iRaftStateContext
	mInitOnce  sync.Once
	mStartOnce sync.Once
	mWinOnce   sync.Once

	// update: init / ceAskingForVote
	mTerm int64

	// update: ceInit / ceCandidateRequestVote / ceAskingForVote / ceVoteToCandidate
	mVotedTerm        int64
	mVotedCandidateID string
	mVotedTimestamp   int64

	// update: ceInit / ceAskingForVote / ceReceiveTicket / ceDisposing
	mTicketCount map[string]bool
	mTicketMutex *sync.Mutex

	// update: ceInit / ceDisposing
	mDisposedFlag bool
}

// trigger: init()
// args: empty
const ceInit = "candidate.init"

// trigger: Start()
// args: empty
const ceStart = "candidate.Start"

// trigger: whenAskingForVoteThenWatchElectionTimeout()
// args: empty
const ceElectionTimeout = "candidate.ElectionTimeout"

// trigger: Heartbeat() / AppendLog() / CommitLog()
// args: empty
const ceLeaderAnnounced = "candidate.LeaderAnnounced"

// trigger: RequestVote()
// args: *rpc.RequestVoteCmd
const ceVoteToCandidate = "candidate.VoteToCandidate"

// trigger: whenLeaderAnnouncedThenSwitchToFollower()
// args: empty
const ceDisposing = "candidate.Disposing"

// trigger: RequestVote()
// args: empty
const ceCandidateRequestVote = "candidate.CandidateRequestVote"

// trigger: beginAskForVote()
// args: empty
const ceAskingForVote = "candidate.AskingForVote"

// trigger: handleRequestVoteOK()
// args: empty
const ceReceiveTicket = "candidate.ReceiveTicket"

// trigger: whenReceiveTicketThenCheckTicketCount
// args: empty
const ceWinningTheVote = "candidate.ceWinningTheVote"

func newCandidateState(ctx iRaftStateContext, term int64) IRaftState {
	it := new(tCandidateState)
	it.init(ctx, term)
	return it
}

func (me *tCandidateState) init(ctx iRaftStateContext, term int64) {
	me.mInitOnce.Do(func() {
		me.context = ctx
		me.mTerm = term
		me.initEventHandlers()

		me.Raise(ceInit)
	})
}

func (me *tCandidateState) initEventHandlers() {
	// write only logic
	me.hookEventsForTerm()
	me.hookEventsForVotedTerm()
	me.hookEventsForTicketCount()
	me.hookEventsForDisposedFlag()

	// read only logic
	me.Hook(ceStart,
		me.whenStartThenAskForVote)
	me.Hook(ceAskingForVote,
		me.whenAskingForVoteThenWatchElectionTimeout)
	me.Hook(ceReceiveTicket,
		me.whenReceiveTicketThenCheckTicketCount)
	me.Hook(ceElectionTimeout,
		me.whenElectionTimeoutThenAskForVoteAgain)
	me.Hook(ceWinningTheVote,
		me.whenWinningTheVoteThenSwitchToLeader)
	me.Hook(ceLeaderAnnounced,
		me.whenLeaderAnnouncedThenSwitchToFollower)

}

// hookEventsForTerm maintains field: mTerm
// update: ceElectionTimeout
func (me *tCandidateState) hookEventsForTerm() {
	me.Hook(ceAskingForVote, func(e string, args ...interface{}) {
		me.mTerm++
		me.logf(".ceAskingForVote, term=%v", me.mTerm)
	})
}

// hookEventsForVotedTerm maintains field: mVotedTerm
// update: ceInit / ceBeforeRequestVote / ceAskingForVote / ceVoteToCandidate
func (me *tCandidateState) hookEventsForVotedTerm() {
	me.Hook(ceInit, func(e string, args ...interface{}) {
		// initially, vote to itself
		me.mVotedTerm = me.mTerm
		me.mVotedCandidateID = me.context.Config().GetID()
		me.mVotedTimestamp = time.Now().UnixNano()
	})

	me.Hook(ceCandidateRequestVote, func(e string, args ...interface{}) {
		// before RequestVote, if last vote timeout then reset vote state
		if time.Duration(time.Now().UnixNano()-me.mVotedTimestamp)*time.Nanosecond >= timeout.ElectionTimeout {
			me.mVotedTerm = 0
			me.mVotedCandidateID = ""
			me.mVotedTimestamp = 0

			me.logf(".ceCandidateRequestVote, reset voted term to empty")
		}
	})

	me.Hook(ceAskingForVote, func(e string, args ...interface{}) {
		// when timeout, reset to itself
		me.mVotedTerm = me.mTerm
		me.mVotedCandidateID = me.context.Config().GetID()
		me.mVotedTimestamp = time.Now().UnixNano()

		me.logf(".ceAskingForVote, vote to myself")
	})

	me.Hook(ceVoteToCandidate, func(e string, args ...interface{}) {
		// after vote to candidate
		cmd := args[0].(*rpc.RequestVoteCmd)
		me.mVotedTerm = cmd.Term
		me.mVotedCandidateID = cmd.CandidateID
		me.mVotedTimestamp = time.Now().UnixNano()

		me.logf(".ceVoteToCandidate, candidate=%v, term=%v", cmd.CandidateID, cmd.Term)
	})
}

func (me *tCandidateState) hookEventsForTicketCount() {
	me.Hook(ceInit, func(e string, args ...interface{}) {
		me.mTicketMutex = new(sync.Mutex)
		me.mTicketCount = make(map[string]bool, 0)
		me.mTicketCount[me.context.Config().GetID()] = true
	})

	me.Hook(ceAskingForVote, func(e string, args ...interface{}) {
		me.mTicketMutex.Lock()
		defer me.mTicketMutex.Unlock()

		me.mTicketCount = make(map[string]bool, 0)
		me.mTicketCount[me.context.Config().GetID()] = true

		me.logf(".ceAskingForVote, ticketCount=%v", len(me.mTicketCount))
	})

	me.Hook(ceReceiveTicket, func(e string, args ...interface{}) {
		peerID := args[0].(string)

		me.mTicketMutex.Lock()
		defer me.mTicketMutex.Unlock()
		me.mTicketCount[peerID] = true

		me.logf(".ceReceiveTicket, mTicketCount=%v", len(me.mTicketCount))
	})

	me.Hook(ceDisposing, func(e string, args ...interface{}) {
		me.mTicketMutex.Lock()
		defer me.mTicketMutex.Unlock()
		me.mTicketCount = make(map[string]bool, 0)

		me.logf(".ceDisposing, mTicketCount=0")
	})
}

func (me *tCandidateState) hookEventsForDisposedFlag() {
	me.Hook(ceInit, func(e string, args ...interface{}) {
		me.mDisposedFlag = false
	})

	me.Hook(ceDisposing, func(e string, args ...interface{}) {
		me.mDisposedFlag = true
		me.logf(".ceDisposing, mDisposedFlag=true")
	})
}

func (me *tCandidateState) Heartbeat(cmd *rpc.HeartbeatCmd, ret *rpc.HeartbeatRet) error {
	// check term
	if cmd.Term <= me.mTerm {
		// bad leader
		ret.Code = rpc.HBTermMismatch
		return nil
	}

	// new leader
	me.Raise(ceLeaderAnnounced, cmd.Term)

	// return ok
	ret.Code = rpc.HBOk
	return nil
}

func (me *tCandidateState) AppendLog(cmd *rpc.AppendLogCmd, ret *rpc.AppendLogRet) error {
	// check term
	if cmd.Term <= me.mTerm {
		// bad leader
		ret.Code = rpc.ALTermMismatch
		return nil
	}

	// new leader
	me.Raise(ceLeaderAnnounced, cmd.Term)

	// ignore and return
	ret.Code = rpc.ALInternalError
	return nil
}

func (me *tCandidateState) CommitLog(cmd *rpc.CommitLogCmd, ret *rpc.CommitLogRet) error {
	// ignore and return
	ret.Code = rpc.CLInternalError
	return nil
}

func (me *tCandidateState) RequestVote(cmd *rpc.RequestVoteCmd, ret *rpc.RequestVoteRet) error {
	me.Raise(ceCandidateRequestVote, cmd)

	// check voted term
	if cmd.Term < me.mVotedTerm {
		ret.Code = rpc.RVTermMismatch
		return nil
	}

	if cmd.Term == me.mVotedTerm {
		if me.mVotedCandidateID != "" && me.mVotedCandidateID != cmd.CandidateID {
			// already vote another
			ret.Code = rpc.RVVotedAnother
			return nil
		} else {
			// already voted
			ret.Code = rpc.RVOk
			return nil
		}
	}

	if cmd.Term > me.mVotedTerm {
		// new term, check log
		if cmd.LastLogIndex >= me.context.Store().LastCommittedIndex() {
			// good log
			me.Raise(ceVoteToCandidate, cmd)
			ret.Code = rpc.RVOk

		} else {
			// bad log
			ret.Code = rpc.RVLogMismatch
		}

		return nil
	}

	// should not reach here
	ret.Code = rpc.RVTermMismatch
	return nil
}

func (me *tCandidateState) Role() roles.RaftRole {
	return roles.Candidate
}

func (me *tCandidateState) Start() {
	me.mStartOnce.Do(func() {
		me.Raise(ceStart)
	})
}

func (me *tCandidateState) whenLeaderAnnouncedThenSwitchToFollower(_ string, args ...interface{}) {
	me.Raise(ceDisposing)

	term := args[0].(int64)
	me.logf(".whenLeaderAnnouncedThenSwitchToFollower, term=%v")
	me.context.HandleStateChanged(newFollowerState(me.context, term))
}

func (me *tCandidateState) whenElectionTimeoutThenAskForVoteAgain(_ string, _ ...interface{}) {
	me.logf(".whenElectionTimeoutThenAskForVoteAgain")
	me.beginAskForVote()
}

func (me *tCandidateState) whenStartThenAskForVote(_ string, _ ...interface{}) {
	me.logf(".whenStartThenAskForVote")
	me.beginAskForVote()
}

func (me *tCandidateState) beginAskForVote() {
	// raise ceAskingForVote
	me.Raise(ceAskingForVote)

	// for each node, call node.RequestVote
	cmd := new(rpc.RequestVoteCmd)
	cmd.CandidateID = me.context.Config().GetID()
	cmd.Term = me.mTerm

	store := me.context.Store()
	cmd.LastLogIndex = store.LastCommittedIndex()
	cmd.LastLogTerm = store.LastCommittedTerm()

	term := me.mTerm
	for _, node := range me.context.Config().GetNodes() {
		if node.GetID() == me.context.Config().GetID() {
			continue
		}

		peerID := node.GetID()
		go func() {
			ret := new(rpc.RequestVoteRet)
			err := me.context.RaftClientService().UsingRaft(peerID, func(client rpc.IRaftRPC) error {
				return client.RequestVote(cmd, ret)
			})

			if err == nil && ret.Code == rpc.RVOk {
				me.handleRequestVoteOK(peerID, term)
			}
		}()
	}
}

func (me *tCandidateState) whenAskingForVoteThenWatchElectionTimeout(_ string, _ ...interface{}) {
	me.logf(".whenAskingForVoteThenWatchElectionTimeout")

	term := me.mTerm
	go func() {
		time.Sleep(timeout.RandElectionTimeout())

		if me.mDisposedFlag || me.mTerm != term {
			return
		}

		tc := me.getTicketCount()
		if tc < len(me.context.Config().GetNodes())/2+1 {
			me.Raise(ceElectionTimeout)
		}
	}()
}

func (me *tCandidateState) handleRequestVoteOK(peerID string, term int64) {
	if me.mDisposedFlag || me.mTerm != term {
		return
	}

	me.logf(".handleRequestVoteOK, peer=%v, term=%v", peerID, term)
	me.Raise(ceReceiveTicket, peerID)
}

func (me *tCandidateState) whenReceiveTicketThenCheckTicketCount(_ string, _ ...interface{}) {
	me.logf(".whenReceiveTicketThenCheckTicketCount")

	me.mTicketMutex.Lock()
	defer me.mTicketMutex.Unlock()

	tc := me.getTicketCount()
	if tc >= len(me.context.Config().GetNodes())/2+1 {
		// win the vote
		me.mWinOnce.Do(func() {
			go me.Raise(ceWinningTheVote)
		})
	}
}

func (me *tCandidateState) getTicketCount() int {
	return len(me.mTicketCount)
}

func (me *tCandidateState) whenWinningTheVoteThenSwitchToLeader(_ string, _ ...interface{}) {
	me.logf(".whenWinningTheVoteThenSwitchToLeader")

	me.Raise(ceDisposing)
	me.context.HandleStateChanged(newLeaderState(me.context, me.mTerm))
}

func (me *tCandidateState) ExecuteKVCmd(cmd *rpc.KVCmd, ret *rpc.KVRet) error {
	ret.Code = rpc.KVInternalError
	return gErrorLeaderOffline
}

func (me *tCandidateState) logf(format string, args ...interface{}) {
	prefix := fmt.Sprintf("tCandidateState(%v)", me.context.Config().GetID())
	logger.Logf(prefix+format, args...)
}
